package cn.com.homework.utils;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class IOUtil {

	// read
	// write

	/**
	 * 写字节数组到HDFS文件
	 * 
	 * @param b
	 * @param hdfsPath
	 */
	public static void writeByteArray2HdfsFile(byte[] b, String hdfsPath,
			Configuration conf) {
		// 1. b和hdfsPath是否合法
		FileUtil.checkPath(hdfsPath);

		// 2. 写字节数组到HDFS文件中
		FSDataOutputStream out = null;
		FileSystem fs = null;
		try {
			// 2.1 创建文件
			fs = FileSystem.get(conf);
			// 3. 写文件
			out = fs.create(new Path(hdfsPath)); // new
													// FileOutputStream(outFile)
			out.write(b);

		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				out.close();
				fs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}

	/**
	 * 读取本地文件成字节数组
	 * 
	 * @param localPath
	 * @return
	 */
	public static byte[] readLocalFile2ByteArray(String localPath) {
		// 1. localPath是否合法
		FileUtil.checkPath(localPath);
		// 2. 读取指定路径的文件内容
		FileInputStream fis = null;
		ByteArrayOutputStream bos = null;
		try {
			// 文件输入流
			fis = new FileInputStream(localPath);
			// 字节数组输出流
			bos = new ByteArrayOutputStream();

			// 读写过程
			byte[] bs = new byte[1024];
			int len = 0;
			while ((len = fis.read(bs)) > 0) {
				bos.write(bs, 0, len);
			}

			return bos.toByteArray();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				fis.close();
				bos.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		return null;
	}

	/**
	 * 读取HDFS文件，每一行是一个元素，读取到Set
	 */
	public static Set<String> readHdfsFile2Set(String hdfsPath,
			String charsetName, Configuration conf) {
		// 1. 判断合法性
		FileUtil.checkPath(hdfsPath);

		// 存放数据
		Set<String> result = new HashSet<String>();

		BufferedReader br = null;
		FileSystem fs = null;
		try {
			// 获取文件系统
			fs = FileSystem.get(conf);
			// 打开文件流
			FSDataInputStream in = fs.open(new Path(hdfsPath));
			InputStreamReader isr = new InputStreamReader(in, charsetName);
			br = new BufferedReader(isr);
			// 读数据
			String line = null;
			while ((line = br.readLine()) != null) {
				result.add(line.trim());
			}

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				br.close();
				fs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return result;
	}

	public static List<String> readHdfsFile2List(String hdfsPath,
			String charsetName, Configuration conf) {
		// 1. 判断合法性
		FileUtil.checkPath(hdfsPath);

		// 存放数据
		List<String> result = new ArrayList<String>();

		BufferedReader br = null;
		FileSystem fs = null;
		try {
			// 获取文件系统
			fs = FileSystem.get(conf);
			// 打开文件流
			FSDataInputStream in = fs.open(new Path(hdfsPath));
			InputStreamReader isr = new InputStreamReader(in, charsetName);
			br = new BufferedReader(isr);
			// 读数据
			String line = null;
			while ((line = br.readLine()) != null) {
				result.add(line.trim());
			}

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				br.close();
				fs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return result;
	}

	/**
	 * 读取本地文件到一个set
	 * 
	 * @param localPath
	 * @param charsetName
	 * @return
	 */
	public static Set<String> readLocalFile2Set(String localPath,
			String charsetName) {
		// 1.合法性检验
		FileUtil.checkPath(localPath);

		// 存放内容的Set
		Set<String> result = new HashSet<String>();
		// 创建流
		BufferedReader br = null;
		try {
			FileInputStream fis = new FileInputStream(localPath);
			InputStreamReader isr = new InputStreamReader(fis, charsetName);
			br = new BufferedReader(isr);
			// 读内容
			String line = null;
			while ((line = br.readLine()) != null) {
				result.add(line.trim());
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				br.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return result;
	}

}
